package com.zb.framework.integration;

import com.google.common.collect.Maps;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Bean;
import org.springframework.core.env.Environment;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.stereotype.Component;

import java.util.HashMap;

/**
 * @author fujianjian
 * @date 2023/3/21 16:56
 **/
@Slf4j
@Component
@ConditionalOnProperty(prefix = "zeus.itg.kafka", value = "enable", havingValue = "true")
public class KafkaPerformanceLogListener implements EnvironmentAware {

    private Environment environment;

    @Override
    public void setEnvironment(Environment environment) {
        this.environment = environment;
    }

    @KafkaListener(topics = "zeus-performance-log", containerFactory = "perfKafkaListenerContainerFactory")
    public void onPerformanceMessage(ConsumerRecord<String, String> consumerRecord) {
        log.info("key:{}, \t value:{}", consumerRecord.key(), consumerRecord.value());
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, String> perfKafkaListenerContainerFactory() {
        HashMap<String, Object> configMap = Maps.newHashMap();
        configMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, environment
                .resolvePlaceholders("${zeus.itg.kafka.servers:10.88.3.240:9092,10.88.3.240:9093,10.88.3.240:9094}"));
        configMap.put(ConsumerConfig.GROUP_ID_CONFIG, environment
                .resolvePlaceholders("${zeus.itg.kafka.group-id:stone-zeus-projectname}"));
        configMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        configMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);
        configMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

        ConsumerFactory<String, String> consumerFactory = new DefaultKafkaConsumerFactory<>(configMap);

        ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory();
        factory.setConcurrency(2);
        factory.setConsumerFactory(consumerFactory);
        return factory;
    }

}
